package com.shujia.spark.opt

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object Demo5RePartition {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("cache")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")

    println(s"studentsRDD分区数据：${studentsRDD.getNumPartitions}")

    /**
     * repartition： 对rdd重分区，返回一个新的rdd,  会产生shuffle
     * repartition可以用于增加分区和减少分区，
     * 增加分区可以增加并行度，在资源充足的情况下， 效率更高
     * 减少分区可以减少产生的小文件的数量
     *
     */
    val rePartRDD: RDD[String] = studentsRDD.repartition(10)

    println(s"rePartRDD分区数据：${rePartRDD.getNumPartitions}")

    /**
     * coalesceL 重分区，，可以设置是否产生shuffle
     * 如果指定shuffle为true,可以用于增加分区和减少分区
     * 如果指定shuffle为false，只能用于减少分区
     *
     */

    val coalesceRDD: RDD[String] = rePartRDD.coalesce(100, shuffle = true)
    println(s"coalesceRDD分区数据：${coalesceRDD.getNumPartitions}")


    /**
     * 当处理好的数据需要保存到磁盘的时候，如果产生了很多的小文件，可以使用coalesce合并小文件
     * 合并的标准：保证合并之后的每一个文件的大小在128M左右
     *
     * 比如数据保存的数据是10G, 最好的情况是合并为80个
     *
     * shuffle = false： 不产生shuffle,效率更好
     *
     */

    coalesceRDD
      .coalesce(1, shuffle = false) //合并小我呢见
      .saveAsTextFile("data/coalesce")

  }

}
